package com.example.flink.service;

import org.springframework.stereotype.Service;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import java.util.*;

@Service
public class LogQueryService {
    @Autowired
    private JdbcTemplate jdbcTemplate;

    public Map<String, Object> queryLogs(String query, int page, int size) {
        // 解析query为SQL WHERE子句
        String whereClause = QueryParser.parse(query);
        int offset = (page - 1) * size;
        String sql = "SELECT * FROM cleaned_app_log WHERE " + whereClause +
                " ORDER BY timestamp DESC LIMIT " + size + " OFFSET " + offset;
        List<Map<String, Object>> logs = jdbcTemplate.queryForList(sql);
        // 查询总数
        String countSql = "SELECT count(*) FROM cleaned_app_log WHERE " + whereClause;
        int total = jdbcTemplate.queryForObject(countSql, Integer.class);
        Map<String, Object> result = new HashMap<>();
        result.put("logs", logs);
        result.put("total", total);
        result.put("page", page);
        result.put("size", size);
        return result;
    }

    public Map<String, Object> statsLogs(String query) {
        String whereClause = QueryParser.parse(query);
        // 日志级别分布
        String sqlLevel = "SELECT level, count(*) as cnt FROM cleaned_app_log WHERE " + whereClause + " GROUP BY level";
        // 时间序列（每小时每级别统计）
        String sqlTime = "SELECT formatDateTime(timestamp, '%Y-%m-%d %H:00:00') as hour, level, count(*) as cnt FROM cleaned_app_log WHERE " + whereClause + " GROUP BY hour, level ORDER BY hour, level";
        Map<String, Object> result = new HashMap<>();
        result.put("levelStats", jdbcTemplate.queryForList(sqlLevel));
        // 合并每小时每级别为一行
        List<Map<String, Object>> raw = jdbcTemplate.queryForList(sqlTime);
        Map<String, Map<String, Object>> hourMap = new LinkedHashMap<>();
        for(Map<String, Object> row : raw) {
            String hour = row.get("hour").toString();
            String level = row.get("level").toString();
            long cnt = ((Number)row.get("cnt")).longValue();
            hourMap.putIfAbsent(hour, new LinkedHashMap<>());
            hourMap.get(hour).put("hour", hour);
            hourMap.get(hour).put(level, cnt);
        }
        // 补全所有级别，避免有的小时缺少某级别导致前端折线断裂
        List<String> allLevels = new ArrayList<>();
        for(Map<String, Object> row : raw) {
            String level = row.get("level").toString();
            if(!allLevels.contains(level)) allLevels.add(level);
        }
        for(Map<String, Object> row : hourMap.values()) {
            for(String level : allLevels) {
                if(!row.containsKey(level)) row.put(level, 0L);
            }
        }
        result.put("timeStats", new ArrayList<>(hourMap.values()));
        return result;
    }
} 